package com.zhxy.util.message;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zhxy.util.async.ResultHandler;

/**
 * 为(request->response)的消息模式提供支持<br/>
 * 1.支持消息响应的管线化(pipeline)
 * 2.支持超时处理
 * 
 * @author 刘勉
 * 
 */
public class PipelineManager implements MessageServiceListener {

	private static final Logger logger = LoggerFactory
			.getLogger(PipelineManager.class);

	private MessageService messageNode;

	private AtomicLong totalSend = new AtomicLong(0);

	private AtomicLong totalReceive = new AtomicLong(0);

	private RequestResponseProcessor messageProcessor;

	private volatile boolean closed = false;

	private ScheduledExecutorService executorService;

	private long timeout = 30000;

	public void setMessagePipe(MessageService messagePipe) {
		this.messageNode = messagePipe;
	}

	public void setMessageProcessor(RequestResponseProcessor messageProcessor) {
		this.messageProcessor = messageProcessor;
	}

	public void setExecutorService(ScheduledExecutorService executorService) {
		this.executorService = executorService;
	}

	public void init() {
		messageNode.addMessageServiceListener(this);
		// new Thread(new Worker()).start();

		executorService.scheduleAtFixedRate(new Worker(), 0, 100,
				TimeUnit.MILLISECONDS);
	}

	public void close() {
		closed = true;
		messageNode.removeMessageServiceListener(this);
		
		MessagePipeException e = new MessagePipeException("系统已关闭");
		for (MessagePipe pipe : messageNode.getMessagePipes()) {
			PipelineSession ps = (PipelineSession)pipe.getAttribute(PipelineSession.class.getName());
			ps.close(e);
			pipe.removeAttribute(PipelineSession.class.getName());
		}
	}

	public boolean closed() {
		return closed;
	}

	public void send(MessagePipe pipe, Object message,
			ResultHandler<Object> handler) {
		
		PipelineSession ps = (PipelineSession)pipe.getAttribute(PipelineSession.class.getName());
		if (messageProcessor.isRequest(pipe, message)) {
			// waitForReply.add(handler);
			//waitingQueues.get(pipe).add(handler);
			ps.addRequest(handler);
			pipe.send(message);
		} else {
			pipe.send(message);
			handler.result(null);
		}

		totalSend.incrementAndGet();
	}

	@Override
	public void connected(MessagePipe pipe) {
		
		PipelineSession ps = new PipelineSession();
		ps.setTimeout(timeout);
		pipe.setAttribute(PipelineSession.class.getName(), ps);
	}

	@Override
	public void disconnected(MessagePipe pipe) {
		
		logger.debug("连接关闭，清理资源");

		// 在这里通知队列中中所有等待的对象,连接已断开
		PipelineSession ps = (PipelineSession)pipe.getAttribute(PipelineSession.class.getName());
		ps.close(new MessagePipeException("连接已断开"));

	}

	@Override
	public void message(MessagePipe pipe, Object message) {
		
		PipelineSession ps = (PipelineSession)pipe.getAttribute(PipelineSession.class.getName());

		if (messageProcessor.isResponse(pipe, message)) {
			//ResultHandler<Object> handler = waitingQueues.get(pipe).remove();
			//handler.result(message);
			ps.removeRequest().result(message);
		} else {

			if (messageProcessor.isRequest(pipe, message)) {
				//DelayResultHandler handler = new DelayResultHandler();
				//messageProcessor.process(pipe, message, handler);
				//replyQueues.get(pipe).add(handler);
				messageProcessor.process(pipe, message, ps.newReponseHandler());
			} else {
				messageProcessor.process(pipe, message, null);
			}
		}

		totalReceive.incrementAndGet();
	}

	@Override
	public void keepAlive(MessagePipe connection) {

	}

	class Worker implements Runnable {

		private void check() {
			long now = System.currentTimeMillis();
			for (MessagePipe pipe : messageNode.getMessagePipes()) {
				PipelineSession ps = (PipelineSession)pipe.getAttribute(PipelineSession.class.getName());
				try {
					ps.checkRequestQueueTimeout(now);
				} catch (TimeoutException e) {
					logger.warn("请求管道超时，断开连接");
					pipe.close();
					break;
				}
				try {
					while (true) {
						Object result = ps.checkReponseQueue(now);
						if (result != null) {
							pipe.send(result);
						} else {
							break;
						}
					}
				} catch (TimeoutException e) {
					logger.warn("响应管道超时，断开连接");
					pipe.close();
					break;
				}
			}
		}

		@Override
		public void run() {
			try {
				check();
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}

	}

}
